AWS Secrets Managerでクレデンシャル管理してAzure StorageにAPIを投げてみた

AWS EC2インスタンスからAzure APIをセキュアに呼び出すため、Azure接続文字列をAWS Secrets Managerで管理してみました。

#セキュリティ

#Microsoft Azure

#AWS Secrets Manager

#AWS

quiver

2020.08.13

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

Amazon EC2インスタンスからAWSのサービスを利用する場合、AWS Security Token Service (AWS STS) で一時的なセキュリティ認証情報を取得してセキュアにAPIを呼び出す事ができます。

一方で、AWS外のサービスを利用したい場合、永続的なアクセスキーを払い出してAPI を呼び出すケースが多く、アクセスキーの管理に悩みます。

後者のユースで、AWS Secrets Manager(Azure Key VaultのAWS版)を利用してセキュアに接続文字列を管理し、Azure Storage(Amazon S3のAzure版)にPythonクライアントから通信する機会がありましたので、手順を紹介します。

構成

事前準備

Azure/AWS それぞれのPython SDKをインストールします。

# Azure
$ pip install azure-storage-blob
# AWS
$ pip install boto3

手順1. Azure Storageの接続文字列を AWS Secrets Managerに登録

Azure Storage Account で発行したアクセスキー(接続文字列)を AWS Secrets Manager に登録します。

$ aws secretsmanager create-secret --name dev/azure \
  --secret-string "DefaultEndpointsProtocol=https;AccountName=..."
{
    "ARN": "arn:aws:secretsmanager:ap-northeast-1:1234:secret:dev/azure-ZZZ",
    "Name": "dev/azure",
    "VersionId": "1-2-3-4"
}

シークレットは Name : Secret のキー:バリュー形式で登録します。バリューには、文字列、JSON、バイナリなど、様々なデータ形式を保存できます。

JSON 形式の場合は次のようになります。

{
    "ConnectionString" : "DefaultEndpointsProtocol=https;AccountName=..."
}

$ aws secretsmanager create-secret --name dev/azure \
  --secret-string file://cred.json
{
    "ARN": "arn:aws:secretsmanager:ap-northeast-1:1234:secret:dev/azure-ZZZ",
    "Name": "dev/azure",
    "VersionId": "1-2-3-4"
}

手順2. 呼び出し元EC2にAWS Secrets Managerの参照権限を付与

EC2からAzureへのAPI呼び出しでは永続的なアクセスキーを利用しますが、EC2からAWSへのAPI呼び出しでは、一時認証情報を利用します。そのために、EC2インスタンスにIAMをロールを割り当て、ポリシーでAWS Secrets Manager の権限を付与します。

EC2インスタンス(アプリケーション)からは、シークレットの管理は許可せず、取得のみ行えるようにしたのが次です。

{
   "Version":"2012-10-17",
   "Statement":[
      {
         "Effect":"Allow",
         "Action":"secretsmanager:GetSecretValue",
         "Resource":"arn:aws:secretsmanager:ap-northeast-1:1234:secret:dev/azure-ZZZ"
      }
   ]
}

Resourceにはシークレット作成時に払い出されたキークレットの ARN を指定します。

手順3-a. Python プログラムからシークレットを取得する場合

AWS の secretsmanager::get_secret 関数でAWS Secrets Manager管理した接続文字列を取得し、 Azure のBlobServiceClient.from_connection_stringに渡します。

import boto3
import base64
from botocore.exceptions import ClientError

import os
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient

SECRET_NAME="dev/azure"
CONTAINER_NAME="dummy"

def get_secret():
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager'
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=SECRET_NAME
        )
    except ClientError as e:
        if e.response['Error']['Code'] == 'DecryptionFailureException':
            # Secrets Manager can't decrypt the protected secret text using the provided KMS key.
            # Deal with the exception here, and/or rethrow at your discretion.
            raise e
        elif e.response['Error']['Code'] == 'InternalServiceErrorException':
            # An error occurred on the server side.
            # Deal with the exception here, and/or rethrow at your discretion.
            raise e
        elif e.response['Error']['Code'] == 'InvalidParameterException':
            # You provided an invalid value for a parameter.
            # Deal with the exception here, and/or rethrow at your discretion.
            raise e
        elif e.response['Error']['Code'] == 'InvalidRequestException':
            # You provided a parameter value that is not valid for the current state of the resource.
            # Deal with the exception here, and/or rethrow at your discretion.
            raise e
        elif e.response['Error']['Code'] == 'ResourceNotFoundException':
            # We can't find the resource that you asked for.
            # Deal with the exception here, and/or rethrow at your discretion.
            raise e

    # Decrypts secret using the associated KMS CMK.
    # Depending on whether the secret is a string or binary, one of these fields will be populated.
    return get_secret_value_response['SecretString']

# 接続文字列を取得
connect_str = get_secret()

blob_service_client = BlobServiceClient.from_connection_string(connect_str)
container_client = blob_service_client.get_container_client(CONTAINER_NAME)

# コンテナ内のブロブ名一覧を表示
blob_list = container_client.list_blobs()

for blob in blob_list:
    print(blob.name)

get_secret関数の95％は、Secrets Managerコンソールにあるサンプルコードです。

シークレットがJSON形式の場合、get_secret_value で取得した値は文字列型のため、JSON型に変換します。

>>> get_secret_value_response = client.get_secret_value(SecretId="dev/azure")
>>> get_secret_value_response['SecretString']
'{\n    "ConnectionString" : "..."\n}\n'
>>> import json
>>> secret = json.loads(get_secret_value_response['SecretString'])
>>> secret
{'ConnectionString': '...'}
>>> connect_str = secret['ConnectionString']

手順3-b. 親プログラム(Bash シェル)でシークレットを取得し環境変数で渡す場合

Azureと通信するプログラム(Python"azure_storage_sample_env.py")は、環境変数(AZURE_STORAGE_CONNECTION_STRING)で接続文字列を取得するようになっているかもしれません。そのような場合は、Azureと通信するプログラムの呼び出し元で環境変数を設定しておきます。

呼び出し元が Bash シェルスクリプトの場合は、次のようになります。

Azureと通信するプログラムを呼び出すBashシェルスクリプト

secretsmanager::get-secret-value APIでAWS Secrets Manager管理した接続文字列を取得し、環境変数(AZURE_STORAGE_CONNECTION_STRING)に渡します。

#!/usr/bin/env bash
export AZURE_STORAGE_CONNECTION_STRING=$(
  aws secretsmanager get-secret-value \
    --secret-id dev/azure \
    --query SecretString \
    --output text)
python3 azure_storage_sample_env.py

Azureと通信するプログラム

import os
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient

CONTAINER_NAME="dummy"

connect_str = os.getenv('AZURE_STORAGE_CONNECTION_STRING')

blob_service_client = BlobServiceClient.from_connection_string(connect_str)
container_client = blob_service_client.get_container_client(CONTAINER_NAME)

# コンテナ内のブロブ名一覧を表示
blob_list = container_client.list_blobs()

for blob in blob_list:
    print(blob.name)